package com.cbh.amadeus.client.receiver.register;

import com.cbh.amadeus.client.annotation.AmadeusMessageRegister;
import com.cbh.amadeus.client.config.SubscriberConfig;
import com.cbh.amadeus.client.receiver.service.MessageHolder;
import com.cbh.amadeus.client.utils.MessageHolderProxyChange;
import com.cbh.amadeus.common.enums.MessageType;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.protocol.AmadeusMessageProtocol;
import com.cbh.amadeus.common.protocol.AmadeusMessageWrap;
import com.cbh.amadeus.common.utils.StringUtil;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.AnnotationUtils;

import java.text.Collator;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author ChenBiHui
 * @date 2025/4/28
 */
@Slf4j
public class MessageRegisterProxy {

    private static final Map<Integer, List<MessageHolder<?>>> SORT_WITH_HOLDERS = new ConcurrentHashMap<>();

    public MessageRegisterProxy(SubscriberConfig config, List<MessageHolder<?>> holders) {
        List<MessageHolder<?>> copy = new ArrayList<>(holders);
        int size = copy.size();
        if (size == 0) {
            log.warn("===== Current client not have any [MessageHolder], can't receive any message with topic =====");
            return;
        }
        Collator collator = Collator.getInstance(Locale.ROOT);
        collator.setStrength(Collator.TERTIARY);
        copy.sort(((o1, o2) -> collator.compare(o1.getClass().getName(), o2.getClass().getName())));
        Integer count = config.getChannelCount();
        if (count > size) {
            throw new IllegalArgumentException(
                    "The channel count is [" + count + "] greater than the [MessageHolder] count, " +
                            "please adjust the channel count to less than or equal to the [MessageHolder] count, " +
                            "[MessageHolder] count is [" + copy.size() + "], respective from to " + copy.stream()
                            .map(item -> item.getClass().getName())
                            .collect(Collectors.toList())
            );
        }
        int part = size / count;
        int remain = size % count;
        int compensate = 0;
        int start = 0;
        for (int i = 0; i < count; i++) {
            List<MessageHolder<?>> target = new ArrayList<>();
            SORT_WITH_HOLDERS.put(i, target);
            int end = start + part + (++compensate <= remain ? 1 : 0);
            target.addAll(holders.subList(start, Math.min(end, size)));
            start = end;
        }
    }

    public void register(Channel channel, String name, Integer sort) {
        if (SORT_WITH_HOLDERS.isEmpty()) {
            log.warn("===== Current client not have register any topic to the server =====");
            channel.writeAndFlush(onlyConnect(name));
            return;
        }
        List<MessageHolder<?>> holders = SORT_WITH_HOLDERS.get(sort);
        if (holders == null || holders.isEmpty()) {
            log.warn(
                    "===== register sort [{}] is null or empty =====",
                    sort
            );
            channel.writeAndFlush(onlyConnect(name));
            return;
        }
        for (MessageHolder<?> holder : holders) {
            AmadeusMessageRegister register = AnnotationUtils.findAnnotation(holder.getClass(), AmadeusMessageRegister.class);
            if (register == null) {
                log.warn(
                        "===== Can't find @MessageRegister annotation in class [{}] =====",
                        holder.getClass()
                );
                continue;
            }
            String topic = register.topic();
            if (StringUtil.unHasText(topic)) {
                log.warn(
                        "===== Can't register topic [{}] as @MessageRegister annotation in class [{}] =====",
                        topic,
                        holder.getClass()
                );
                continue;
            }
            channel.writeAndFlush(registering(name, topic, MessageHolderProxyChange.get(holder)));
        }
    }

    private AmadeusMessageWrap onlyConnect(String name) {
        AmadeusMessageWrap wrap = new AmadeusMessageWrap();
        wrap.setMagic(AmadeusMessageProtocol.MAGIC);
        wrap.setType((MessageType.ONLY_CONNECT_SEND.getType().byteValue()));
        wrap.setMessage(
                new AmadeusMessage()
                        .setProvider(name)
        );
        return wrap;
    }

    private AmadeusMessageWrap registering(String name, String topic, String label) {
        AmadeusMessageWrap wrap = new AmadeusMessageWrap();
        wrap.setMagic(AmadeusMessageProtocol.MAGIC);
        wrap.setType((MessageType.REGISTER.getType().byteValue()));
        wrap.setMessage(
                new AmadeusMessage()
                        .setProvider(name)
                        .setTopic(topic)
                        .setPayload(label)
        );
        return wrap;
    }
}
